package cn.piesat.client.service.impl;

import cn.piesat.client.constant.cache.QsCache;
import cn.piesat.client.constant.consist.AppProperties;
import cn.piesat.client.constant.enums.ClientRecordEnum;
import cn.piesat.client.converter.QsTaskStatusConverter;
import cn.piesat.client.model.dto.buiness.QsClientRunnerDto;
import cn.piesat.client.model.vo.QsTaskStatusVo;
import cn.piesat.client.service.DefaultClientRecord;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class QsClientRecord extends DefaultClientRecord<QsClientRunnerDto> {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @Autowired
    private AppProperties appProperties;

    @Autowired
    private QsTaskStatusConverter qsTaskStatusConverter;

    @Value("${spring.kafka.producer.state.topic}")
    private String statusTopic;

    @Override
    public void doRunRecord(QsClientRunnerDto qsClientRunnerDto) {
        QsTaskStatusVo qsTaskStatusVo = qsTaskStatusConverter.task2RunStatusVo(qsClientRunnerDto, appProperties.getIp());
        if(qsTaskStatusVo!=null){
            QsCache.taskStatusCache.put(qsClientRunnerDto.getFileTaskId(),qsTaskStatusVo);
            kafkaTemplate.send(statusTopic,JSON.toJSONString(qsTaskStatusVo, SerializerFeature.WriteNullStringAsEmpty));
        }
    }

    @Override
    public void deEndRecord(QsClientRunnerDto qsClientRunnerDto, String resultJson) {
        QsTaskStatusVo qsTaskStatusVo = QsCache.taskStatusCache.get(qsClientRunnerDto.getFileTaskId());
        if(qsTaskStatusVo!=null){
            JSONObject algorithmResultsJson = null;
            ClientRecordEnum clientRecordEnum = null;
            if(resultJson!=null){
                JSONObject jsonObject = JSON.parseObject(resultJson);
                algorithmResultsJson = jsonObject.getJSONObject("algorithmResultsJson");
                if(algorithmResultsJson != null){
                    String status = algorithmResultsJson.getString("status");
                    clientRecordEnum = ClientRecordEnum.getClientRecordEnumOfResultNumber(status);
                }
            }else{
                clientRecordEnum = ClientRecordEnum.UNKNOWN;
            }
            qsTaskStatusVo = qsTaskStatusConverter.task2EndStatusVo(qsTaskStatusVo, clientRecordEnum);
            kafkaTemplate.send(statusTopic,JSON.toJSONString(qsTaskStatusVo, SerializerFeature.WriteNullStringAsEmpty));
        }
    }
}
